package io.kiki.sba.registry.server.metadata.remoting.notifier;

import io.kiki.sba.registry.api.Callback;
import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.api.exchange.NodeExchanger;
import io.kiki.sba.registry.api.exchange.message.Response;
import io.kiki.sba.registry.api.exchange.message.SimpleRequest;
import io.kiki.sba.registry.common.model.Node;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.metaserver.ProvideDataChangeEvent;
import io.kiki.sba.registry.common.model.metaserver.RemoteDatumClearEvent;
import io.kiki.sba.registry.common.model.metaserver.SlotsChangeEvent;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.common.model.store.Url;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.remoting.connection.NodeConnectManager;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.DefaultExecutorFactory;
import io.kiki.sba.registry.util.OsUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;


public abstract class AbstractNotifier<T extends Node> implements Notifier {

    protected final Logger logger = LoggerFactory.getLogger(getClass());

    @Autowired
    protected MetadataLeaderService metadataLeaderService;

    private Executor executors = DefaultExecutorFactory.createCachedThreadPoolFactory(getClass().getSimpleName(), Math.min(4, OsUtils.getCpuCount()), 60 * 1000, TimeUnit.MILLISECONDS).create();

    @Override
    public void notifySlotsChange(Slots slots) {
        if (metadataLeaderService.amIStableAsLeader()) {
            new NotifyTemplate<SlotsChangeEvent>().broadcast(new SlotsChangeEvent(slots.getEpoch()));
        }
    }

    @Override
    public void notifyProvideDataChange(ProvideDataChangeEvent event) {
        new NotifyTemplate<ProvideDataChangeEvent>().broadcast(event);
    }

    @Override
    public void notifyRemoteDatumClear(RemoteDatumClearEvent event) {
        if (metadataLeaderService.amIStableAsLeader()) {
            new NotifyTemplate<RemoteDatumClearEvent>().broadcast(event);
        }
    }

    public Map<String, Object> broadcastInvoke(Object request, int timeout) throws Exception {
        return new InvokeTemplate().broadcast(request, timeout);
    }


    public AbstractNotifier<T> setExecutors(Executor executors) {
        this.executors = executors;
        return this;
    }

    protected abstract NodeExchanger getNodeExchanger();

    protected abstract List<T> getDataStoreNodeList();

    protected abstract NodeConnectManager getNodeConnectManager();

    private Tuple<Set<String>, Collection<InetSocketAddress>> getNodeConnections() {
        NodeConnectManager nodeConnectManager = getNodeConnectManager();
        Collection<InetSocketAddress> connections = nodeConnectManager.getChannelRemoteAddressList(null);

        if (connections == null || connections.isEmpty()) {
            logger.error("Push Node list error! No node connected!");
            return null;
        }

        List<T> nodes = getDataStoreNodeList();

        if (nodes == null || nodes.isEmpty()) {
            logger.error("Node list error! No node registered!");
            return null;
        }
        Set<String> ipAddresses = new HashSet<>();
        nodes.forEach(node -> ipAddresses.add(node.getUrl().getIp()));
        return new Tuple<>(ipAddresses, connections);
    }


    public final class InvokeTemplate<R> {
        public Map<String, Object> broadcast(R req, int timeout) throws Exception {
            Tuple<Set<String>, Collection<InetSocketAddress>> tuple = getNodeConnections();
            if (tuple == null) {
                return new HashMap<>();
            }
            Set<String> ipAddresses = tuple.getFirst();
            Collection<InetSocketAddress> connections = tuple.getSecond();

            Map<String, Object> map = new ConcurrentHashMap<>();
            boolean succeed = new ConcurrentUtils.SafeParaLoop<InetSocketAddress>(executors, connections) {
                @Override
                protected void doRun0(InetSocketAddress connection) throws Exception {
                    try {
                        String address = connection.getAddress().getHostAddress();
                        if (!ipAddresses.contains(address)) {
                            return;
                        }
                        if (map.containsKey(address)) {
                            return;
                        }
                        Response resp = getNodeExchanger().request(new SimpleRequest(req, new Url(connection)));
                        if (resp != null) {
                            map.put(address, resp.getResult());
                        } else {
                            logger.error("broadcast request {} to {} failed: response null", req, connection.getAddress());
                        }
                    } catch (Throwable e) {
                        logger.error("broadcast request {} to {} failed:", req, connection.getAddress(), e);
                    }
                }
            }.runAndWait(timeout);
            if (!succeed) {
                logger.error("broadcast request {} failed:", req);
            }
            return map;
        }
    }

    public final class NotifyTemplate<E> {

        public void broadcast(E event) {
            Tuple<Set<String>, Collection<InetSocketAddress>> nodeConnections = getNodeConnections();
            if (nodeConnections == null) {
                return;
            }

            Set<String> ipAddresses = nodeConnections.getFirst();
            Collection<InetSocketAddress> connections = nodeConnections.getSecond();

            new ConcurrentUtils.SafeParaLoop<InetSocketAddress>(executors, connections) {
                @Override
                protected void doRun0(InetSocketAddress connection) throws Exception {
                    if (!ipAddresses.contains(connection.getAddress().getHostAddress())) {
                        return;
                    }
                    getNodeExchanger().request(new NotifyRequest(event, connection, executors));
                }
            }.run();
        }
    }

    private final class NotifyRequest<E> extends SimpleRequest<E> {
        private final Executor executors;

        public NotifyRequest(E event, InetSocketAddress connection, Executor executors) {
            super(event, new Url(connection));
            this.executors = executors;
        }

        @Override
        public Callback getCallBack() {
            return new Callback() {
                @Override
                public void onSuccess(Channel channel, Object message) {
                    logger.info("[onCallback] notify slot-change succeed, ({}): [{}]", channel != null ? channel.getRemoteAddress() : "unknown channel", message);
                }

                @Override
                public void onFailed(Channel channel, Throwable exception) {
                    logger.error("[onException] notify slot-change failed, ({})", channel != null ? channel.getRemoteAddress() : "unknown channel", exception);
                }

                @Override
                public Executor getExecutor() {
                    return executors;
                }
            };
        }
    }
}
